package net.bwie.realtime.jtp.dws.douyin.log.functions;

import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.OrderSourceMetric3;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class OrderSourceRatioWindowFunction3 implements WindowFunction<
        EventLog6,                // 输入数据：过滤后的下单事件
        OrderSourceMetric3,           // 输出数据：占比指标
        Long,                               // 分组键：主播ID（anchorId）
        TimeWindow> {                      // 时间窗口

    @Override
    public void apply(
            Long anchorId,                  // 分组键：主播ID
            TimeWindow window,              // 窗口信息（开始/结束时间）
            Iterable<EventLog6> input,  // 窗口内所有下单事件
            Collector<OrderSourceMetric3> out) throws Exception {

        // 1. 初始化统计容器：key=来源类型，value=该来源订单数
        Map<String, Long> sourceOrderCountMap = new HashMap<>();
        long totalOrderCount = 0L;  // 窗口内总订单数

        // 2. 遍历窗口数据，统计各来源订单数和总订单数
        Iterator<EventLog6> iterator = input.iterator();
        while (iterator.hasNext()) {
            EventLog6 event = iterator.next();
            String sourceType = event.getSourceType();

            // 累加该来源订单数（不存在则初始化为1，存在则+1）
            sourceOrderCountMap.put(sourceType, sourceOrderCountMap.getOrDefault(sourceType, 0L) + 1);
            // 累加总订单数
            totalOrderCount++;
        }

        // 3. 格式化窗口时间（转换为yyyy-MM-dd HH:mm:ss）
        long windowStart = window.getStart();
        long windowEnd = window.getEnd();
        String windowStartTime = DateTimeUtil.convertLongToString(windowStart, "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(windowEnd, "yyyy-MM-dd HH:mm:ss");
        String curDate = windowStartTime.substring(0, 10);  // 提取日期

        // 4. 计算各来源占比并输出指标
        for (Map.Entry<String, Long> entry : sourceOrderCountMap.entrySet()) {
            String sourceType = entry.getKey();
            long orderCount = entry.getValue();
            // 计算占比（避免总订单为0导致除0错误）
            double orderRatio = totalOrderCount == 0 ? 0.0 : Math.round((orderCount * 100.0 / totalOrderCount) * 100) / 100.0;

            // 封装指标并输出
            OrderSourceMetric3 metric = new OrderSourceMetric3(
                    windowStartTime,
                    windowEndTime,
                    curDate,
                    anchorId,
                    sourceType,
                    orderCount,
                    totalOrderCount,
                    orderRatio
            );
            out.collect(metric);
        }
    }
}
